{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "import torch.nn as nn\n",
    "import torch.optim as optim\n",
    "\n",
    "class Agent(nn.Module):\n",
    "    def __init__(self, input_size, action_size):\n",
    "        super(Agent, self).__init__()\n",
    "        self.fc1 = nn.Linear(input_size, 128)\n",
    "        self.fc2 = nn.Linear(128, action_size)\n",
    "    \n",
    "    def forward(self, x):\n",
    "        x = torch.relu(self.fc1(x))\n",
    "        return torch.softmax(self.fc2(x), dim=-1)  # 选择动作\n",
    "\n",
    "class MultiAgentSystem:\n",
    "    def __init__(self):\n",
    "        self.text_agent = Agent(input_size=10, action_size=2)  # 文本智能体\n",
    "        self.audio_agent = Agent(input_size=10, action_size=2)  # 声学智能体\n",
    "        self.optimizer = optim.Adam(list(self.text_agent.parameters()) + list(self.audio_agent.parameters()))\n",
    "    \n",
    "    def step(self, text_data, audio_data):\n",
    "        text_action = self.text_agent(text_data)  # 文本智能体选择动作\n",
    "        audio_action = self.audio_agent(audio_data)  # 声学智能体选择动作\n",
    "        reward = self.calculate_reward(text_action, audio_action)  # 计算奖励\n",
    "        \n",
    "        # 更新智能体的策略\n",
    "        self.optimizer.zero_grad()\n",
    "        loss = -reward  # 假设我们要最大化奖励\n",
    "        loss.backward()\n",
    "        self.optimizer.step()\n",
    "    \n",
    "    def calculate_reward(self, text_action, audio_action):\n",
    "        # 一个简单的奖励机制\n",
    "        if torch.argmax(text_action) == torch.argmax(audio_action):\n",
    "            return 1  # 如果两个智能体协作良好，给予奖励\n",
    "        else:\n",
    "            return -1  # 否则给予惩罚"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "\n",
    "class Agent:\n",
    "    def __init__(self, name):\n",
    "        self.name = name\n",
    "        self.policy = self.init_policy()\n",
    "    \n",
    "    def init_policy(self):\n",
    "        # 初始化随机策略（例如：一个简单的二分类任务：保留或去除词汇）\n",
    "        return np.random.rand()\n",
    "\n",
    "    def choose_action(self, state):\n",
    "        # 基于当前状态和策略，选择一个动作\n",
    "        if np.random.rand() > self.policy:\n",
    "            return \"retain\"  # 保留词汇\n",
    "        else:\n",
    "            return \"remove\"  # 去除词汇\n",
    "\n",
    "    def update_policy(self, reward):\n",
    "        # 根据奖励调整策略\n",
    "        self.policy += reward * 0.1  # 简单的策略更新规则\n",
    "\n",
    "# 定义多智能体环境\n",
    "class MultiAgentEnvironment:\n",
    "    def __init__(self):\n",
    "        self.text_agent = Agent(\"Text Agent\")\n",
    "        self.acoustic_agent = Agent(\"Acoustic Agent\")\n",
    "        self.global_reward = 0\n",
    "\n",
    "    def step(self, state):\n",
    "        # 每个智能体在其模态中选择动作\n",
    "        text_action = self.text_agent.choose_action(state)\n",
    "        acoustic_action = self.acoustic_agent.choose_action(state)\n",
    "        \n",
    "        # 简单的奖励机制（这里可以根据实际情感分类任务的结果来给出反馈）\n",
    "        reward = self.evaluate(text_action, acoustic_action)\n",
    "        \n",
    "        # 更新智能体的策略\n",
    "        self.text_agent.update_policy(reward)\n",
    "        self.acoustic_agent.update_policy(reward)\n",
    "        \n",
    "        return reward\n",
    "\n",
    "    def evaluate(self, text_action, acoustic_action):\n",
    "        # 评估两个智能体的协作结果，例如两个模态的动作是否有助于分类\n",
    "        if text_action == \"retain\" and acoustic_action == \"retain\":\n",
    "            return 1  # 奖励高\n",
    "        elif text_action == \"remove\" and acoustic_action == \"remove\":\n",
    "            return -1  # 惩罚\n",
    "        else:\n",
    "            return 0  # 中性反馈\n",
    "\n",
    "# 初始化环境并进行训练\n",
    "env = MultiAgentEnvironment()\n",
    "state = \"input_data\"\n",
    "\n",
    "for episode in range(100):  # 模拟训练100个回合\n",
    "    reward = env.step(state)\n",
    "    print(f\"Episode {episode}: Reward: {reward}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2. **深度强化学习**\n",
    "\n",
    "在深度强化学习中，智能体通过神经网络来学习一个策略，这个策略能帮助它在不同状态下做出最佳动作。下面是一个简单的**Q-learning**（一种常见的强化学习算法）示例，结合神经网络来优化策略：\n",
    "#### **解释**：\n",
    "\n",
    "- **DQN 类**：一个简单的三层全连接神经网络，负责学习如何从状态预测动作的价值（Q值）。\n",
    "- **DQNAgent 类**：智能体通过`DQN`模型选择动作，并通过强化学习算法更新其策略。它会根据经验回放（经验记忆池中的历史状态、动作、奖励和下一个状态）来训练模型。\n",
    "- **折扣因子和探索率**：折扣因子（`gamma`）用来决定未来奖励的影响程度，探索率（`epsilon`）用来平衡探索（选择随机动作）和利用（选择最优动作）。\n",
    "\n",
    "### 总结\n",
    "\n",
    "- **多智能体协作**：多个智能体一起工作，它们分别处理不同模态的数据，共同优化任务。\n",
    "- **深度强化学习**：通过神经网络，智能体能够从环境中学习，选择最优的动作来最大化长期奖励。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "import torch.nn as nn\n",
    "import torch.optim as optim\n",
    "import numpy as np\n",
    "\n",
    "# 定义一个简单的深度Q网络\n",
    "class DQN(nn.Module):\n",
    "    def __init__(self, input_size, output_size):\n",
    "        super(DQN, self).__init__()\n",
    "        self.fc1 = nn.Linear(input_size, 128)\n",
    "        self.fc2 = nn.Linear(128, 64)\n",
    "        self.fc3 = nn.Linear(64, output_size)\n",
    "    \n",
    "    def forward(self, x):\n",
    "        x = torch.relu(self.fc1(x))\n",
    "        x = torch.relu(self.fc2(x))\n",
    "        return self.fc3(x)\n",
    "\n",
    "# 定义智能体\n",
    "class DQNAgent:\n",
    "    def __init__(self, input_size, output_size):\n",
    "        self.model = DQN(input_size, output_size)\n",
    "        self.optimizer = optim.Adam(self.model.parameters(), lr=0.001)\n",
    "        self.criterion = nn.MSELoss()  # 使用均方误差作为损失函数\n",
    "        self.memory = []  # 经验回放\n",
    "        self.gamma = 0.95  # 折扣因子\n",
    "        self.epsilon = 0.1  # 探索率\n",
    "\n",
    "    def act(self, state):\n",
    "        if np.random.rand() < self.epsilon:\n",
    "            return np.random.randint(0, 2)  # 随机选择动作\n",
    "        else:\n",
    "            with torch.no_grad():\n",
    "                return torch.argmax(self.model(torch.Tensor(state))).item()  # 选择最优动作\n",
    "\n",
    "    def remember(self, state, action, reward, next_state):\n",
    "        self.memory.append((state, action, reward, next_state))\n",
    "\n",
    "    def train(self):\n",
    "        if len(self.memory) < 100:  # 确保有足够的经验数据\n",
    "            return\n",
    "        batch = np.random.choice(self.memory, 32)  # 随机取32条经验\n",
    "        for state, action, reward, next_state in batch:\n",
    "            q_values = self.model(torch.Tensor(state))\n",
    "            next_q_values = self.model(torch.Tensor(next_state))\n",
    "            target = q_values.clone()\n",
    "            target[action] = reward + self.gamma * torch.max(next_q_values).item()\n",
    "            \n",
    "            loss = self.criterion(q_values, target)\n",
    "            self.optimizer.zero_grad()\n",
    "            loss.backward()\n",
    "            self.optimizer.step()\n",
    "\n",
    "# 模拟训练过程\n",
    "input_size = 10  # 假设状态空间维度为10\n",
    "output_size = 2  # 动作空间有两个动作：保留或移除\n",
    "\n",
    "agent = DQNAgent(input_size, output_size)\n",
    "\n",
    "# 简单的环境和状态转换\n",
    "for episode in range(100):  # 训练100个回合\n",
    "    state = np.random.rand(input_size)  # 假设状态是随机生成的\n",
    "    action = agent.act(state)\n",
    "    reward = np.random.rand()  # 假设奖励也是随机生成的\n",
    "    next_state = np.random.rand(input_size)\n",
    "    \n",
    "    agent.remember(state, action, reward, next_state)\n",
    "    agent.train()\n",
    "    print(f\"Episode {episode}: Action: {action}, Reward: {reward}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "import torch.nn as nn\n",
    "import torch.optim as optim\n",
    "\n",
    "class Agent(nn.Module):\n",
    "    def __init__(self, input_size, action_size):\n",
    "        super(Agent, self).__init__()\n",
    "        self.fc1 = nn.Linear(input_size, 128)\n",
    "        self.fc2 = nn.Linear(128, action_size)\n",
    "    \n",
    "    def forward(self, x):\n",
    "        x = torch.relu(self.fc1(x))\n",
    "        return torch.softmax(self.fc2(x), dim=-1)  # 选择动作\n",
    "\n",
    "class MultiAgentSystem:\n",
    "    def __init__(self):\n",
    "        self.text_agent = Agent(input_size=10, action_size=2)  # 文本智能体\n",
    "        self.audio_agent = Agent(input_size=10, action_size=2)  # 声学智能体\n",
    "        self.optimizer = optim.Adam(list(self.text_agent.parameters()) + list(self.audio_agent.parameters()))\n",
    "    \n",
    "    def step(self, text_data, audio_data):\n",
    "        text_action = self.text_agent(text_data)  # 文本智能体选择动作\n",
    "        audio_action = self.audio_agent(audio_data)  # 声学智能体选择动作\n",
    "        reward = self.calculate_reward(text_action, audio_action)  # 计算奖励\n",
    "        \n",
    "        # 更新智能体的策略\n",
    "        self.optimizer.zero_grad()\n",
    "        loss = -reward  # 假设我们要最大化奖励\n",
    "        loss.backward()\n",
    "        self.optimizer.step()\n",
    "    \n",
    "    def calculate_reward(self, text_action, audio_action):\n",
    "        # 一个简单的奖励机制\n",
    "        if torch.argmax(text_action) == torch.argmax(audio_action):\n",
    "            return 1  # 如果两个智能体协作良好，给予奖励\n",
    "        else:\n",
    "            return -1  # 否则给予惩罚"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 示例代码\n",
    "\n",
    "下面的代码展示了如何先**单独训练每个智能体**，然后再将它们的预训练模型加载到多智能体环境中进行进一步训练："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "import torch.nn as nn\n",
    "import torch.optim as optim\n",
    "\n",
    "# 定义单个智能体模型（文本或声学）\n",
    "class SimpleAgent(nn.Module):\n",
    "    def __init__(self, input_size, output_size):\n",
    "        super(SimpleAgent, self).__init__()\n",
    "        self.fc1 = nn.Linear(input_size, 128)\n",
    "        self.fc2 = nn.Linear(128, output_size)\n",
    "    \n",
    "    def forward(self, x):\n",
    "        x = torch.relu(self.fc1(x))\n",
    "        return self.fc2(x)\n",
    "\n",
    "# 训练单个智能体的函数\n",
    "def train_single_agent(agent, data_loader, epochs=10):\n",
    "    optimizer = optim.Adam(agent.parameters(), lr=0.001)\n",
    "    loss_fn = nn.CrossEntropyLoss()\n",
    "    \n",
    "    for epoch in range(epochs):\n",
    "        for data, target in data_loader:\n",
    "            output = agent(data)\n",
    "            loss = loss_fn(output, target)\n",
    "            optimizer.zero_grad()\n",
    "            loss.backward()\n",
    "            optimizer.step()\n",
    "        print(f\"Epoch {epoch+1}/{epochs}, Loss: {loss.item()}\")\n",
    "    \n",
    "    return agent\n",
    "\n",
    "# 初始化单个智能体并训练\n",
    "text_agent = SimpleAgent(input_size=10, output_size=2)  # 文本智能体\n",
    "audio_agent = SimpleAgent(input_size=10, output_size=2)  # 声学智能体\n",
    "\n",
    "# 模拟数据加载器\n",
    "# 假设 data_loader 是你的训练数据的迭代器\n",
    "# data_loader = ...\n",
    "\n",
    "# 分别训练文本智能体和声学智能体\n",
    "text_agent = train_single_agent(text_agent, data_loader)  # 训练文本智能体\n",
    "audio_agent = train_single_agent(audio_agent, data_loader)  # 训练声学智能体\n",
    "\n",
    "# --- 预训练完成后，将它们实例化到多智能体系统中 ---\n",
    "\n",
    "class MultiAgentSystem:\n",
    "    def __init__(self, text_agent, audio_agent):\n",
    "        self.text_agent = text_agent\n",
    "        self.audio_agent = audio_agent\n",
    "        self.optimizer = optim.Adam(list(self.text_agent.parameters()) + list(self.audio_agent.parameters()))\n",
    "    \n",
    "    def step(self, text_data, audio_data):\n",
    "        text_action = self.text_agent(text_data)  # 文本智能体选择动作\n",
    "        audio_action = self.audio_agent(audio_data)  # 声学智能体选择动作\n",
    "        reward = self.calculate_reward(text_action, audio_action)  # 计算奖励\n",
    "        \n",
    "        # 更新智能体的策略\n",
    "        self.optimizer.zero_grad()\n",
    "        loss = -reward  # 假设我们要最大化奖励\n",
    "        loss.backward()\n",
    "        self.optimizer.step()\n",
    "\n",
    "    def calculate_reward(self, text_action, audio_action):\n",
    "        # 一个简单的奖励机制\n",
    "        if torch.argmax(text_action) == torch.argmax(audio_action):\n",
    "            return 1  # 如果两个智能体协作良好，给予奖励\n",
    "        else:\n",
    "            return -1  # 否则给予惩罚\n",
    "\n",
    "# 实例化多智能体系统，并将预训练好的模型加载\n",
    "multi_agent_system = MultiAgentSystem(text_agent, audio_agent)\n",
    "\n",
    "# 现在你可以在多智能体环境中进一步训练\n",
    "for episode in range(100):\n",
    "    # 模拟文本和音频数据输入\n",
    "    text_data = torch.randn(10)  # 假设输入是随机生成的\n",
    "    audio_data = torch.randn(10)\n",
    "    \n",
    "    multi_agent_system.step(text_data, audio_data)\n",
    "    print(f\"Episode {episode+1}: Completed\")"
   ]
  }
 ],
 "metadata": {
  "language_info": {
   "name": "python"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
